//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
/////////////////////////////////////////////////////////// FOR CUTTING INPUT INTO LOGICAL PARTS /////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

package trunk.hadoop;

import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapred.RecordReader;

import java.io.IOException;

public class daRecordReader implements RecordReader<IntWritable, BytesWritable> {
    private int    position,   blockEndOffset,  blockSize;
    private boolean continueI = true;
    private daInputFileFetcher fetch;

    // Constructor
    public daRecordReader(FSDataInputStream inputStream, int blockStartOffset, int blockSize, int longestQueryLength, int mapBlockSize)
    {
        fetch = new daInputFileFetcher(inputStream, blockStartOffset, blockSize, longestQueryLength, mapBlockSize);
        position = blockStartOffset;
        this.blockSize = mapBlockSize - longestQueryLength;
        blockEndOffset = blockStartOffset + blockSize - longestQueryLength - 1;

    }
    @Override
    public boolean next(IntWritable key, BytesWritable value) throws IOException {

        if(!continueI)  return false;

        key.set(position);
        if(fetch.read(value) == -1) continueI = false;

        position += (blockSize);  // Each call of this method, ONLY return that position is shifted by ##    // for real shift go to input fetcher
        return true;
    }
    @Override
    public IntWritable createKey()         { return new IntWritable(); }
    @Override
    public BytesWritable createValue()      { return new BytesWritable();}
    @Override
    public long getPos() throws IOException { return position; }
    @Override
    public void close() throws IOException  { fetch.close(); }
    @Override
    public float getProgress() throws IOException { return ( position/blockEndOffset )*100; }
}
